package com.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql._
import org.apache.spark.sql.expressions.Aggregator

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

/**
  *
  * @author Pop
  * @date 2022/7/20 22:48
  */
object Spark08_SparkSql_Test2 {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root")
    // 创建SparkSql的运行环境
    val sparkSql: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSql")
    val session: SparkSession = SparkSession.builder()
      .enableHiveSupport()//启动hive支持，很重要
      .config(sparkSql).getOrCreate()
    // 查询基本数据
    session.sql("use popbase")

    /**
     * 由于在需求中，我们需要获得点击的比率，这个是方法是sparksql中没有的，所以要
     * 自定义udaf函数了
     *
     * 我们需要获得某个商品在某个城市的点击比率，那么就需要获得总数量，和单独的某个
     * 商品的数量，然后进行相除。
     */
    session.sql(
      """
        |select
        | a.*,
        | p.product_name, --商品名称
        | c.city_name, -- 城市名称
        | c.area -- 地区
        | from
        | user_visit_action a
        | join product_info  p on p.product_id = a.click_product_id
        | join city_info c on c.city_id = a.city_id
        | where a.click_product_id > -1 --是点击操作
        |""".stripMargin).createOrReplaceTempView("t1")
    // 创建一个零食表 t1
      // 根据区域和商品进行分组聚合
      session.udf.register("cityRemark",functions.udaf(new CityRemarkUDAF))
      session.sql(
        """
          | select
          | area,product_name,
          | count(*) as clickCnt, --点击的次数
          | cityRemark(city_name) as city_remark --这里，我们需要定义一个udaf的函数，来获取备注
          | from
          | t1 group by area,product_name -- 按照地区和产品名分组
          |""".stripMargin).createOrReplaceTempView("t2")

    // 区域内，对点击数量进行排行
    session.sql(
      """
        | select *,-- 以区域进行排序，按照点击次数，降序
        |    rank() over(partition by area order by clickCnt desc) as rank
        |    from t2
        |""".stripMargin).createOrReplaceTempView("t3")
    //取前三名
    session.sql(
      """
        |select * from t3 where rank <=3
        |""".stripMargin).show(false)
    // 由于备注很长，所以默认会自动截取，变成... 我们不希望截取，所以使用false参数
    session.close()
  }

  case class Buffer(var total:Long,var cityMap:mutable.Map[String,Long])
  /**
   * 自定义聚合函数，实现城市备注功能
   *
   * 1.继承 Aggregator 自定泛型
   *
   *   IN:城市名称
   *   BUF: 【总点击数量，Map(城市，点击数量)】
   *   OUT: 备注信息
   *
   */
  class CityRemarkUDAF extends Aggregator[String,Buffer,String]{

    // 缓冲区的初始化
    override def zero: Buffer = {
      Buffer(0,mutable.Map[String,Long]())
    }
    // 更新缓冲区
    override def reduce(b: Buffer, a: String): Buffer = {
      b.total += 1 // 累计总数
      val num:Long = b.cityMap.getOrElse(a,0L)+1//使用cityName来找Map里的数量，默认是0
      b.cityMap.update(a,num)
      b
    }
    // 涉及分布式计算，所以要合并集合
    override def merge(b1: Buffer, b2: Buffer): Buffer = {
      b1.total +=b2.total
      val map1=b1.cityMap
      val map2 = b2.cityMap
      b1.cityMap = map1.foldLeft(map2){
        case (map,(key,value))=>{
          val c =map.getOrElse(key,0L) + value
          map.update(key,c)
          map
        }
      }
      b1
    }
    // 返回结果,生成字符串信息
    override def finish(reduction: Buffer): String = {
      val remarkList = ListBuffer[String]()
      //拿出数据
      val totalCnt = reduction.total
      val cityMap = reduction.cityMap

      //首先是排序，由于map是无法排序的，所以这里将其转换为List
      val cityCntList: List[(String, Long)] =
        cityMap.toList.sortWith(
        (left, right) => {
          left._2 > right._2 //用map的数量做排序
        }
      ).take(2)//取两个
      // 是否还有更多
      val hasMore = cityMap.size > 2

      //计算所占比率
      var rsum = 0L
      cityCntList.foreach{
        case (city, cnt) => {
          val r = cnt * 100 /totalCnt
          remarkList.append(s"${city} ${r}%")
          rsum += r
        }
      }

      if(hasMore){
        remarkList.append(s"其他 ${100-rsum}%")
      }

      remarkList.mkString(",")
    }

    override def bufferEncoder: Encoder[Buffer] = Encoders.product

    override def outputEncoder: Encoder[String] = Encoders.STRING
  }

}
